package com.atguigu.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.common.Constant;
import com.atguigu.util.HBaseUtil;
import com.atguigu.util.JedisUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

public class DimSinkFunction extends RichSinkFunction<JSONObject> {

    private Connection connection;
    private Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = HBaseUtil.getConnection();
        jedis = JedisUtil.getJedis();
    }

    //value:{"database":"gmall-220623-flink","table":"comment_info","type":"insert","ts":1669162958,"xid":1111,"xoffset":13941,"data":{"id":1595211185799847960,"user_id":119,"nick_name":null,"head_img":null,"sku_id":31,"spu_id":10,"order_id":987,"appraise":"1204","comment_txt":"评论内容：48384811984748167197482849234338563286217912223261","create_time":"2022-08-02 08:22:38","operate_time":null},"sink_table":"dim_comment_info","row_key_column":"id","sink_extend":null}
    @Override
    public void invoke(JSONObject value, Context context) throws Exception {

        //获取数据操作类型
        String type = value.getString("type");
        String sinkTable = value.getString("sink_table");
        JSONObject data = value.getJSONObject("data");
        String rowKeyColumn = value.getString("row_key_column");
        String rowKey = data.getString(rowKeyColumn);
        //判断预分区键是否为null
        String sinkExtend = value.getString("sink_extend");
        if (sinkExtend != null) {
            rowKey = HBaseUtil.getRowKey(rowKey, sinkExtend);
        }

        if ("delete".equals(type)) {
            HBaseUtil.deleteData(connection, Constant.HBASE_NAME_SPACE, sinkTable, rowKey);
            JedisUtil.deleteData(jedis, sinkTable, rowKey);
        } else {
            data.remove(rowKeyColumn);

            //如果为更新操作,则将更新数据直接同步到Redis中  保证Redis与HBase数据存储的一致性
            if ("update".equals(type)) {
                JedisUtil.setData(jedis, sinkTable, rowKey, data.toJSONString());
            }

            HBaseUtil.putJsonData(connection, Constant.HBASE_NAME_SPACE, sinkTable, rowKey, value.getString("sink_column_family"), data);
        }
    }

    @Override
    public void close() throws Exception {
        connection.close();
    }
}
